package com.myCompany.edu.realtime.util;


import com.alibaba.fastjson.JSONObject;

import com.myCompany.edu.realtime.annotation.NoNeedSink;
import com.myCompany.edu.realtime.bean.TableProcess;
import com.myCompany.edu.realtime.common.Constant;
import com.myCompany.edu.realtime.sink.PhoenixSink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.Properties;



/**
 * @author
 * @date 2022/9/17 21:53
 */

public class FlinkSinkUtil {
    public static void main(String[] args) {

    }

    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {
        return new PhoenixSink();
    }

    public static SinkFunction<String> getKafkaSink(String topic) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", Constant.KAFKA_BROKERS);
        //使用严格一次需要加一个参数
        props.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "");

        return new FlinkKafkaProducer<String>(
                "default",
                new KafkaSerializationSchema<String>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(String element,
                                                                    @Nullable Long timestamp) {
                        return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
                    }
                },
                props,
                FlinkKafkaProducer.Semantic.AT_LEAST_ONCE
        );
    }

    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", Constant.KAFKA_BROKERS);
        //严格一次
        props.setProperty("transaction.timeout.ms", 15 * 60 * 60 + "");
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
                "default",
                new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> element, @Nullable Long timestamp) {
                        String sinkTable = element.f1.getSinkTable();
                        byte[] value = element.f0.toJSONString().getBytes(StandardCharsets.UTF_8);
                        return new ProducerRecord<>(sinkTable, value);
                    }
                },
                props,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }


    public static <T> SinkFunction<T> getClickHouseSink(String table, Class<T> tClass) {
        String driver = "com.clickhouse.jdbc.ClickHouseDriver";
        String url = "jdbc:clickhouse://hadoop102:8123/default";

        List<String> names = AtguiguUtil.getClassFeildsName(tClass);

        StringBuilder sql = new StringBuilder();
        sql
                .append("insert into ")
                .append(table)
                .append("(")
                .append(String.join(",", names))
                .append(") values(")

                /*
                 * 使用 , 作为 names 字符串得分隔符
                 * @param  delimiter a sequence of characters that is used to separate each
                 *         of the {@code elements} in the resulting {@code String}
                 * @param  elements an {@code Iterable} that will have its {@code elements}
                 *         joined together.
                 */

                .append(String.join(",", names).replaceAll("[^,]+", "?"))
                .append(")");
        System.out.println("clickhouse 的插入语句： " + sql);


        return getJdbcSink(driver, url, "default", "123321", sql.toString());
    }

    private static <T> SinkFunction<T> getJdbcSink(String driver,
                                                   String url,
                                                   String user,
                                                   String password,
                                                   String sql) {
        return JdbcSink.sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps, T t) throws SQLException {

                        //TODO 给占位符进行赋值。需要参考sql语句的拼写
                        //通过t.getClass()获取t的表类型
                        Class<?> tClass = t.getClass();
                        //通过getDeclareFields()获取表中的字段
                        Field[] fields = tClass.getDeclaredFields();

                        try {
                            for (int i = 0, position = 1; i < fields.length; i++) {
                                Field field = fields[i];

                                NoNeedSink noNeedSink = field.getAnnotation(NoNeedSink.class);
                                if (noNeedSink == null) {
                                    //开启类型私有属性的赋值权限
                                    field.setAccessible(true);
                                    /*
                                     *get方法
                                     * @param obj object from which the represented field's value is
                                     * to be extracted
                                     * @return the value of the represented field in object
                                     * 从t中获取field这个字段的数据
                                     */
                                    ps.setObject(position++, field.get(t));
                                }

                            }
                        } catch (IllegalAccessException e) {
                            throw new RuntimeException(e);
                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withMaxRetries(3)
                        .withBatchSize(1024 * 1024)
                        .withBatchIntervalMs(2000)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(driver)
                        .withUrl(url)
                        .withUsername(user)
                        .withPassword(password)
                        .build()
        );
    }


}
